package com.atguigu.bigdata.chapter02;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author lzc
 * @Date 2022/8/31 9:51
 */
public class Flink02_WC_UnBounded_Lambda {
    public static void main(String[] args) {
        
        // 1. 创建流的执行环境  自动根据当前运行的环境获取合适的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 通过环境从数据源(source)获取一个流
        DataStreamSource<String> source = env.socketTextStream("hadoop162", 9999);
        // 3. 对流做各种转换
        SingleOutputStreamOperator<String> wordStream = source.flatMap((String line, Collector<String> out) -> {
            for (String word : line.split(" ")) {
                out.collect(word);  // 把单词放入到后序的流中
            }
        }).returns(Types.STRING);
    
        // 泛型擦除
        // 泛型只存在于源码中, 编译成字节码之后泛型是不存在, 这就叫泛型擦除. 在java中正常!!!!
        // 在flink中如果由于泛型擦除的存在导致flink无法知道流中的数据的类型, 则不允许
        SingleOutputStreamOperator<Tuple2<String, Long>> wordOneStream = wordStream
            .map( word -> Tuple2.of(word, 1L))
            .returns(Types.TUPLE(Types.STRING, Types.LONG)); // 明确指定类型
        
        
        // 返回每个元素的key
        KeyedStream<Tuple2<String, Long>, String> keyedStream = wordOneStream
            .keyBy(t -> t.f0);  // 只在这个方法内使用
        // 对元组中某个位置进行聚合
        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = keyedStream.sum(1);
        
        
        // 4. 把流输出(sink)
        resultStream.print();
        
        // 5. 执行 执行环境
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*

Spark:
1. 创建 SparkContext

2. 通过上下对象获取 rdd

3. 对rdd做各种转换算子

4. 执行一个行动算子你

5. 启动上下文


 */